package org.kth.dks.test_bed;

import junit.framework.TestCase;

import org.kth.dks.DKSObject;
import org.kth.dks.dks_comm.ConnectionManager;
import org.kth.dks.dks_comm.DKSOverlayAddress;
import org.kth.dks.dks_comm.DKSRef;
import org.kth.dks.dks_dht.DKSDHTCallback;
import org.kth.dks.dks_dht.DKSDHTImpl;
import org.kth.dks.dks_marshal.DKSMessage;

public class RouteAsyncTest extends TestCase {
	private final int START_PORT = 4440;
	private DKSDHTImpl node1 = null;
	private DKSDHTImpl node2 = null;
	private ConnectionManager cm1 = null;
	private ConnectionManager cm2 = null;
	private AppHandler app1 = null;
	private AppHandler app2 = null;
	private final long ROUTECOUNT = 4000;
	
	public void testManyPing() {
		Thread t1 = new Thread() {
			public void run() {
				for (long cc = 0; cc<ROUTECOUNT; cc++) {
					node1.routeAsync(10, new DKSObject(new PingMsg(cc).flatten()));
					Thread.yield();
				}
			}
		};
		
		Thread t2 = new Thread() {
			public void run() {
				for (long cc = 0; cc<ROUTECOUNT; cc++) {
					node2.routeAsync(0, new DKSObject(new PingMsg(cc).flatten()));
					Thread.yield();
				}
			}
		};
		
	    t1.setName(RouteAsyncTest.class.getName());
	    t2.setName(RouteAsyncTest.class.getName());

		t1.start();
		t2.start();
		try {
			t1.join();
			t2.join();
		} catch (Exception ex) {
			throw new Error(ex);
		}
		
		app1.waitUntilBoolIsTrue();
		app2.waitUntilBoolIsTrue();
		
		long c1 = app1.getCount();
		long c2 = app2.getCount();
		assertEquals(c1, c2);
		assertEquals(c1, ROUTECOUNT);
	}
	
	public void setUp() {
		try {
			DKSMessage.addMessageType(PingMsg.NAME, "org.kth.dks.test_bed.PingMsg");
			cm1 = ConnectionManager.newInstance( (int) START_PORT); // Static factory only used for testing, should use getInstance (singleton)
			cm2 = ConnectionManager.newInstance( (int) START_PORT+1); // Static factory only used for testing, should use getInstance (singleton)
			
			DKSOverlayAddress nodeId1 = new DKSOverlayAddress("dksoverlay://0/0/0");
			DKSOverlayAddress nodeId2 = new DKSOverlayAddress("dksoverlay://0/10/0");
			
			app1 = new AppHandler("node1", ROUTECOUNT);
			app2 = new AppHandler("node2", ROUTECOUNT);
			node1 = new DKSDHTImpl(cm1, nodeId1, app1);
			node2 = new DKSDHTImpl(cm2, nodeId2, app2);
			
			node1.create();
			
			node2.join( DKSRef.valueOf(node1.getDKSURL()));
			
		}
		catch (Exception ex) {
			ex.printStackTrace();
		}		
	}
	
	public void tearDown() {
		sleep(500);
		node1.leave(); // causes the node to gracefully leave
		sleep(500);
		node2.leave(); // causes the node to gracefully leave
		sleep(500);
		
		cm1.end(); // Closes down all the sockets and threads
		cm2.end(); // Closes down all the sockets and threads
	}
	
	public void sleep(int msec) {
		try { 
			Thread.sleep(msec); 
		} 
		catch (Exception ex) 
		{ 
			ex.printStackTrace(); 
		}
	}
	
	public class AppHandler implements DKSDHTCallback {
		private String name;
		private boolean bol;
		private long count;
		private final long ROUTECOUNT;
		
		public synchronized String getName() { return name; }
		public synchronized long getCount() { return count; }
		public synchronized void incCount() { count++; }
		
		public synchronized void waitUntilBoolIsTrue() {
			while (!bol) {
				try {
					wait();
				} catch (InterruptedException ex) {
					ex.printStackTrace();
				}
			}
		}
		
		public synchronized boolean getBool() {
			return bol;	
		}
		
		public synchronized void setBool(boolean b) {
			bol = b;
			if (b) 
				notify();
		}
		
		public void dhtRouteCallbackAsync(long identifier, DKSMessage payload) {
			if (payload instanceof PingMsg) {
				incCount();
				if (getCount()==this.ROUTECOUNT)
					setBool(true);
				PingMsg p = (PingMsg) payload;
				System.out.println(getName()+": got ping("+p.getCount()+")"+" total("+getCount()+")");
			} else {
				System.err.println("Error parsting : "+payload);
			}
		}
		
		public AppHandler(String s, long RC) { 
			synchronized (this) {
				name = s; 
				bol = false;
				count = 0;
				this.ROUTECOUNT = RC;
			}
		}

		public void dhtBroadcastCallback(DKSObject value) {}
		public DKSMessage dhtRouteCallback(long identifier, DKSMessage value)  { return null; }
	}
	
}
